package com.hzgj.bcl.soa.rpc.server;

import com.google.common.collect.Maps;

import com.hzgj.bcl.soa.ChainStack;
import com.hzgj.bcl.soa.rpc.MessageDispatcher;
import com.hzgj.bcl.soa.rpc.RpcContext;
import com.hzgj.bcl.soa.rpc.exception.RpcException;
import com.hzgj.bcl.soa.rpc.exception.RpcInfo;
import com.hzgj.bcl.soa.rpc.message.Message;
import com.hzgj.bcl.soa.rpc.message.Payload;
import com.hzgj.bcl.soa.spi.Server;
import com.hzgj.bcl.soa.stats.StatsCollectors;
import com.hzgj.bcl.soa.stats.StatsData;
import com.hzgj.bcl.soa.util.Constants;
import com.hzgj.bcl.util.jmx.annotation.MBean;
import com.hzgj.bcl.util.jmx.annotation.ManagedOperation;

import java.util.Map;
import java.util.concurrent.atomic.LongAdder;

/**
 * Created by dell on 2015/4/5.
 */
@MBean(description = "RpcServer")
public class RpcServer extends MessageDispatcher implements Server {

    private final MethodReflectContainer methodReflectContainer = new MethodReflectContainer();
    private final StatsCollectors collectors = StatsCollectors.getInstance();
    /**
     * 流量计数器
     */
    private final LongAdder longAdder = new LongAdder();
    ServerConfig serverConfig;
    Map<String, Integer> serviceCounter = Maps.newHashMap();

    public RpcServer(ChainStack stack, ServerConfig serverConfig) {
        super(stack);
        this.serverConfig = serverConfig;
    }

    @ManagedOperation(description = "number of biz count")
    public long getBizCount() {
        return longAdder.longValue();
    }

    @ManagedOperation(description = "get count of service")
    public long getServiceCount(String service) {
        Integer count = serviceCounter.get(service);
        if (count == null) {
            return 0;
        }
        return count;
    }

    @Override
    public void start() throws Exception {
        stack.insertInterceptorAtBottom(new NettyServer(serverConfig));
        super.start();
    }

    @Override
    public Object handle(Message message) throws Exception {
        if (message == null) {
            if (logger.isErrorEnabled()) {
                logger.error("message or message buffer is null");
            }
            return null;
        }

        Object object = message.getPayload();
        if (!(object instanceof Payload)) {
            String protocol = (String) message.getHeader(Constants.MESSAGE_HEADER_PROTOCOL);
            throw new IllegalArgumentException("message contain a invalidate object,protocol by " + protocol);
        }

        Payload payload = (Payload) object;

        MethodExecutor methodExecutor = methodReflectContainer.getExecutor(payload.getService(), payload.getMethod());
        if (methodExecutor == null) {
            throw new RpcException(RpcInfo.SERVER_SERVICE_NOT_FOUND, payload.getService(), payload.getMethod());
        }
        Object[] args = payload.getArgs(methodExecutor.getParameterTypes());
        if (logger.isDebugEnabled()) {
            logger.debug("service:{},method:{},args:{}", payload.getService(), payload.getMethod(), args);
        }

        RpcContext.set(payload.getContextId(), payload.getClientName(), payload.getService() + "_" + payload.getMethod());
        long start = System.currentTimeMillis();
        try {
            longAdder.increment();
            serviceCounter.compute(payload.getService(), (key, value) -> {
                if (value == null) {
                    return 1;
                }
                return ++value;
            });
            Object result = methodExecutor.invoke(args);
            report(message, payload, null, start);
            if (logger.isDebugEnabled()) {
                logger.debug("invoke {} with response {}", payload, result);
            }
            return result;
        } catch (Exception e) {
            report(message, payload, e.getMessage(), start);
            throw e;
        } finally {
            longAdder.decrement();
            serviceCounter.compute(payload.getService(), (key, value) -> {
                if (value == null) {
                    return 0;
                }
                return --value;
            });
        }
    }

    public void registerProcessor(String instanceName, Object instance) {
        Class<?> clazz = null;
        try {
            clazz = Thread.currentThread().getContextClassLoader().loadClass(instanceName);
        } catch (ClassNotFoundException e) {
            logger.error("{}", e);
        }
        methodReflectContainer.registerMethods(clazz, instance, instanceName);
    }

    private void report(Message message, Payload payload, String error, long startTime) {
        StatsData data = new StatsData();
        data.messageId = message.getMessageId();
        data.contextId = payload.getContextId();
        data.parentId = payload.getClientName();
        data.service = payload.getService();
        data.method = payload.getMethod();
        data.success = error == null;
        data.error = error;
        data.time = System.currentTimeMillis();
        data.span = data.time - startTime;
        data.dstHost = (String) message.getHeader(Constants.MESSAGE_HEADER_SERVER_HOST);
        data.dstPort = (Integer) message.getHeader(Constants.MESSAGE_HEADER_SERVER_PORT);
        data.srcHost = (String) message.getHeader(Constants.MESSAGE_HEADER_REMOTE_HOST);
        collectors.invoke(data);
    }

    @Override
    public String getProtocol() {
        return Constants.DEFAULT_PROTOCOL;
    }
}
